package com.wuwangfu.process;

import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.ProcessFunction;
import org.apache.flink.util.Collector;

/**
 * @Author: jcshen
 * @Date: 2023-03-08
 * @PackageName: com.wuwangfu.process
 * @ClassName: ProcessFunction
 * @Description:
 * @Version: 1.0.0
 *
 * https://nightlies.apache.org/flink/flink-docs-release-1.14/docs/dev/datastream/operators/process_function/#process-function
 *
 * ProcessFunctoin过程函数
 *    ProcessFunction(过程函数)是一种底层低级别的流处理操作函数，可以访问所有(无环的)流式应用程序底层
 *    基础的方法和属性。相比于上层高级别的算子，ProcessFunction的使用相对复杂，但是灵活度更高，
 *    ProcessFunction的特点：
 *      1、可以在Non-Keyed Stream和KeyedStream使用ProcessFunction处理流式事件数据；
 *      2、可以在KeyedStream中使用ProcessFunction并使用state实现容错和保证数据一致性
 *      3、可以在KeyedStream中使用ProcessFunction并注册Processing Time和EventTime定时器
 *
 */
public class NonKeyedProcessFunction {
    public static void main(String[] args) throws Exception {

        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        DataStreamSource<String> line = env.socketTextStream("localhost", 8888);

        //对NonKeyed DataStream调用process
        SingleOutputStreamOperator<Tuple2<String, Integer>> wordAndOne = line.process(new ProcessFunction<String, Tuple2<String, Integer>>() {
            @Override
            public void processElement(String value, Context context, Collector<Tuple2<String, Integer>> out) throws Exception {
                String[] words = value.split(" ");
                for (String word : words) {
                    if (!"error".equals(word)) {
                        out.collect(Tuple2.of(word, 1));
                    }
                }
            }
        });
        wordAndOne.print();


        env.execute();
    }
}
